#include "pch.h"
#include "database_thread.h"
#include "sql_operation.h"
#include "query_callback.h"

namespace engine
{
	DatabaseThread::DatabaseThread() : mStop(0), mSqlCond(mSqlMutex)
	{
		mPushSqlQueue = new SqlQueue();
		mPopSqlQueue = new SqlQueue();
		mPushResultQueue = new ResultQueue();
		mPopResultQueue = new ResultQueue();
	}

	DatabaseThread::~DatabaseThread()
	{
		SAFE_DELETE(mPushSqlQueue);
		SAFE_DELETE(mPopSqlQueue);
		SAFE_DELETE(mPushResultQueue);
		SAFE_DELETE(mPopResultQueue);
	}

	void* DatabaseThread::entry()
	{
		while (true)
		{
			mSqlCond.wait();

			bool hasData = true;

			while (hasData)
			{
				{
					CriticalSectionLocker lock(mSqlCS);
					std::swap(mPushSqlQueue, mPopSqlQueue);
				}

				while (!mPopSqlQueue->empty())
				{
					SqlOperation* op = mPopSqlQueue->front();
					op->execute(this);
					delete op;
					mPopSqlQueue->pop();
				}

				hasData = false;

				{
					CriticalSectionLocker lock(mSqlCS);
					hasData = !mPushSqlQueue->empty();
				}
			}

			if (mStop)
				break;
		}

		return 0;
	}

	void DatabaseThread::start()
	{
		create();
		run();
	}

	void DatabaseThread::stop()
	{
		mStop = 1;
		mSqlCond.signal();
		waitExit();
	}

	uint32 DatabaseThread::getQueueSize()
	{
		CriticalSectionLocker lock(mSqlCS);
		return (uint32)mPushSqlQueue->size();
	}

	bool DatabaseThread::push(SqlOperation* op)
	{
		{
			CriticalSectionLocker lock(mSqlCS);
			mPushSqlQueue->push(op);
		}

		mSqlCond.signal();

		return true;
	}

	void DatabaseThread::pushResult(IQueryCallback* callback)
	{
		CriticalSectionLocker lock(mResultCS);
		mPushResultQueue->push(callback);
	}

	void DatabaseThread::handleResults()
	{
		{
			CriticalSectionLocker lock(mResultCS);
			std::swap(mPushResultQueue, mPopResultQueue);
		}

		while (!mPopResultQueue->empty())
		{
			IQueryCallback* cb = mPopResultQueue->front();
			cb->execute();
			delete cb;
			mPopResultQueue->pop();
		}
	}
}